package com.owen.spark.mysql

import org.apache.spark.sql.SparkSession

object Mysql {
  def main(args: Array[String]): Unit = {
    val spark=SparkSession.builder().appName("test").master("local").getOrCreate()
    val monitorPoor=spark.read
      .format("jdbc")
      .option("url","jdbc:mysql://172.22.1.151:13306/ffp_cloud_rsync?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=false&serverTimezone=GMT%2B8&allowMultiQueries=true")
      .option("dbtable","monitor_member_sync")
      .option("user","ceo")
      .option("password","Xczx@2021")
      .load()
    // jdbcDF.show()
    monitorPoor.createOrReplaceTempView("m")

    val tmpMonitorPoor=spark.read
      .format("jdbc")
      .option("url","jdbc:mysql://172.22.1.151:13306/ffp_cloud_rsync?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=false&serverTimezone=GMT%2B8&allowMultiQueries=true")
      .option("dbtable","tmp_monitor_member")
      .option("user","ceo")
      .option("password","Xczx@2021")
      .load()
    tmpMonitorPoor.createOrReplaceTempView("tm")

//    val joinExpression=monitorPoor.col("poor_cardid")===tmpMonitorPoor.col("POOR_CARDID")
//    monitorPoor.join(tmpMonitorPoor,joinExpression).select(monitorPoor("poor_name")).show()

//    spark.sql("select * from m join tm on m.member_cardid<>tm.member_cardid and m.member_no=tm.member_id").show()
    val data=spark.sql("select m.member_no from m join tm on m.member_cardid<>tm.member_cardid and m.member_no=tm.member_id")

    data.repartition(1).foreach(row=>{
      val d=row.getString(0)
      println(d)
    })
  }
}
